package com.disruptor.config;

import com.disruptor.consumer.BaseConsumer;
import com.disruptor.entity.TranslatorDataWrapper;
import com.disruptor.exception.IntEventExceptionHandler;
import com.disruptor.factory.impl.DefaultEventFactoryImpl;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;

/**
 * @author zl
 * @version 1.0
 * @date 2021/10/8 13:35
 */
public class DisruptorAutoConfiguration {

    private int ringBuffer = 1024*1024;


    /**
     * 创建disruptor
     * @param baseConsumer
     * @param events 是否可以重复消费 true 可以 false 不可以 默认不可以
     * @return
     */
    public RingBuffer<TranslatorDataWrapper> createDisruptor(BaseConsumer[] baseConsumer,Boolean events){

        Disruptor disruptor = new Disruptor(new EventFactory() {
            @Override
            public Object newInstance() {
                return new TranslatorDataWrapper();
            }
        }, ringBuffer, Executors.defaultThreadFactory(), ProducerType.SINGLE, new BlockingWaitStrategy());
        //自定义异常
        disruptor.setDefaultExceptionHandler(new IntEventExceptionHandler());
        if(events){
            //允许重复消费
            disruptor.handleEventsWith(baseConsumer);
        }else{
            //不允许重复消费
            disruptor.handleEventsWithWorkerPool(baseConsumer);
        }
        disruptor.start();

        RingBuffer<TranslatorDataWrapper> ringBuffer = disruptor.getRingBuffer();

        return ringBuffer;
    }

    public static void main(String[] args) {
        DisruptorAutoConfiguration disruptorUtils = new DisruptorAutoConfiguration();
        DefaultEventFactoryImpl consumerEventFactory = new DefaultEventFactoryImpl("ceshi");
        BaseConsumer baseConsumer = consumerEventFactory.createConsumer();
        DefaultEventFactoryImpl defaultEventFactory = new DefaultEventFactoryImpl("ceshi2");
        BaseConsumer baseConsumer2 = defaultEventFactory.createConsumer();
        BaseConsumer[] baseConsumers = new BaseConsumer[2];
        baseConsumers[0] = baseConsumer;
        baseConsumers[1] = baseConsumer2;
        RingBuffer<TranslatorDataWrapper> ringBuffer = disruptorUtils.createDisruptor(baseConsumers,true);
        for (int i = 0; i < 10; i++) {
            long sequence = ringBuffer.next();
            TranslatorDataWrapper translatorDataWrapper = ringBuffer.get(sequence);
            translatorDataWrapper.setKey("ceshi"+i);
            Map<String,Object> maps = new HashMap<>();
            maps.put("ceshi",i);
            translatorDataWrapper.setMaps(maps);
            ringBuffer.publish(sequence);
        }
    }
}
